package com.zhw.interceptor;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.nio.charset.Charset;
import java.util.List;
import java.util.Map;

public class TableNameAndTimestampInterceptor implements Interceptor {

    //初始化
    public void initialize() {

    }

    //{"table":"...","database":"..","type":"..","ts":..., "data":{....}}
    //针对单条数据处理
    public Event intercept(Event event) {

        //获取数据
        String json = new String(event.getBody(), Charset.forName("utf-8"));

        JSONObject obj = JSON.parseObject(json);

        String tableName = obj.getString("table");

        long ts = obj.getLong("ts") * 1000;

        Map<String, String> headers = event.getHeaders();

        headers.put("tableName", tableName);

        headers.put("timestamp", ts + "");

        return event;
    }

    //针对一个批次数据处理
    public List<Event> intercept(List<Event> list) {

        for (Event event : list) {
            intercept(event);
        }
        return list;
    }

    //关闭
    public void close() {

    }

    public static class Builder implements org.apache.flume.interceptor.Interceptor.Builder {
        //获取自定义拦截器对象
        @Override
        public Interceptor build() {
            return new TableNameAndTimestampInterceptor();
        }

        @Override
        public void configure(Context context) {

        }
    }

}
